package ex.datastream.physicalPartition;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;

public class Chain {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        FilterFunction<Long> dummyFilter =
                new FilterFunction<Long>() {
                    @Override
                    public boolean filter(Long value) {
                        return false;
                    }
                };

        DataStream<Long> src1 = env.generateSequence(1, 10);
        DataStream<Long> src2 = env.generateSequence(1, 10).slotSharingGroup("src-1");

        // this should not inherit group "src-1"
        src1.union(src2).filter(dummyFilter).printToErr();

        DataStream<Long> src3 = env.generateSequence(1, 10).slotSharingGroup("group-1");
        DataStream<Long> src4 = env.generateSequence(1, 10).slotSharingGroup("group-1");

        // this should inherit "group-1" now
        src3.union(src4).filter(dummyFilter).print();

        JobGraph jobGraph = env.getStreamGraph().getJobGraph();
        List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();

        jobGraph.getVertices().forEach(a->{
            System.out.println(a.getSlotSharingGroup().toString());
        });
//        System.out.println(JSON.toJSON(vertices));

    }
}
